﻿#include "libcomm.h"
#if (COMM_FEATURE_LIBEVENT==1)

#include <event2/event.h>
#include <event2/http.h>
#include <event2/buffer.h>
#include <event2/util.h>
#include <event2/keyvalq_struct.h>
#include <event2/thread.h>


#if (COMM_FEATURE_SSL==1)
#include <openssl/ssl.h>
#include <openssl/err.h>
#include <openssl/rand.h>
#include <event2/bufferevent_ssl.h>

static void SSLError(const char *func)
{
	LOGERR("%s failed.", func);

	/* This is the OpenSSL function that prints the contents of the
	* error stack to the specified file handle. */
	ERR_print_errors_fp(stderr);
}

static SSL_CTX* SSLInit(const char* crtfile, const char* keyfile)
{
	SSL_CTX* ctx;

	SSL_library_init();
	OpenSSL_add_all_algorithms();
	ERR_load_crypto_strings();
	SSL_load_error_strings();
	if (!RAND_poll())
	{
		SSLError("RAND_poll");
		return NULL;
	}

	ctx = SSL_CTX_new(SSLv23_server_method());
	if (NULL == ctx)
	{
		SSLError("SSL_CTX_new");
		return NULL;
	}

	if (NULL != keyfile && NULL != crtfile)
	{
		if (!SSL_CTX_use_certificate_file(ctx, crtfile, SSL_FILETYPE_PEM) ||
			!SSL_CTX_use_PrivateKey_file(ctx, keyfile, SSL_FILETYPE_PEM))
		{
			SSLError("SSL_CTX_use_PrivateKey_file");
			SSL_CTX_free(ctx);
			return NULL;
		}
	}

	//SSL_CTX_set_options(ctx, SSL_OP_NO_SSLv2);
	return ctx;
}

/**
* This callback is responsible for creating a new SSL connection
* and wrapping it in an OpenSSL bufferevent.  This is the way
* we implement an https server instead of a plain old http server.
*/
static struct bufferevent* http_bevcb(struct event_base *base, void *arg)
{
	struct bufferevent* r;
	SSL_CTX *ctx = (SSL_CTX *)arg;

	r = bufferevent_openssl_socket_new(base,
		-1,
		SSL_new(ctx),
		BUFFEREVENT_SSL_ACCEPTING,
		BEV_OPT_CLOSE_ON_FREE);
	return r;
}

#endif
static void http_request_cb(struct evhttp_request *req, void *arg);
static void http_reply_cb(evutil_socket_t sock, short which, void *arg);

class HttpServerImp : public Thread
{
public:
	class RequestItem : public TaskItem
	{
	public:
		struct evhttp_request* m_req;
		HttpServerImp* m_instance;
		bool m_reply_direct;
		RequestItem(struct evhttp_request* req, HttpServerImp* instance, bool reply_direct = false)
			: m_req(req), m_instance(instance), m_reply_direct(reply_direct) {}
		virtual CommType Do()
		{
			HttpServer* server = m_instance->m_server;
			const char *cmdtype;
			const char* uri;
			struct evkeyvalq *headers;

			uri = evhttp_request_get_uri(m_req);
			switch (evhttp_request_get_command(m_req))
			{
			case EVHTTP_REQ_GET: cmdtype = "GET"; break;
			case EVHTTP_REQ_POST: cmdtype = "POST"; break;
			case EVHTTP_REQ_HEAD: cmdtype = "HEAD"; break;
			case EVHTTP_REQ_PUT: cmdtype = "PUT"; break;
			case EVHTTP_REQ_DELETE: cmdtype = "DELETE"; break;
			case EVHTTP_REQ_OPTIONS: cmdtype = "OPTIONS"; break;
			case EVHTTP_REQ_TRACE: cmdtype = "TRACE"; break;
			case EVHTTP_REQ_CONNECT: cmdtype = "CONNECT"; break;
			case EVHTTP_REQ_PATCH: cmdtype = "PATCH"; break;
			default: cmdtype = "unknown"; break;
			}

			StringList requestHeaders;
			{
				struct evkeyval *header;
				headers = evhttp_request_get_input_headers(m_req);
				for (header = headers->tqh_first; header;
					header = header->next.tqe_next) {
					requestHeaders << String::format("%s:%s", header->key, header->value);
				}
			}

			String request;
			{
				struct evbuffer *buf_input;
				buf_input = evhttp_request_get_input_buffer(m_req);
				int content_length = evbuffer_get_length(buf_input);
				if (content_length > 0)
				{
					request.resize(content_length + 1);
					evbuffer_remove(buf_input, (char*)request, content_length);
				}
			}

			Buffer response;
			StringList responseHeaders;
			bool pageFound = server->m_handler->OnRequest(uri, cmdtype, requestHeaders, request, responseHeaders, response);

			headers = evhttp_request_get_output_headers(m_req);
			{
				if (responseHeaders.count() == 0)
				{
					evhttp_add_header(headers, "Content-type", server->m_content_type);
				}
				else
				{
					for (int i = 0; i < responseHeaders.count(); i++)
					{
						StringList kvPair = String::split(responseHeaders[i], ":");
						MYLIB_ASSERT(2 == kvPair.count());
						evhttp_add_header(headers, kvPair[0], kvPair[1]);
					}
				}
			}


			struct evbuffer *buf_output;
			{
				buf_output = evhttp_request_get_output_buffer(m_req);
				if (pageFound)
				{
					evbuffer_add(buf_output, response.get(), response.size());
					if (m_reply_direct)
						evhttp_send_reply(m_req, 200, "OK", NULL);
				}
				else
				{
					if (m_reply_direct)
						evhttp_send_reply(m_req, 404, "not found", NULL);
				}
			}

			if (!m_reply_direct)
				m_instance->PostReplyRequest(m_req, pageFound);

			return CommType::Null;
		}
	};

private:
	event_base* m_base;
	evhttp* m_http;
	event* m_ev;
	HttpServer* m_server;
	int m_index;
	Task m_task;
	ThreadLock m_lock_list;
	typedef std::list< std::pair<evhttp_request*,bool> > ReplyList;
	ReplyList m_reply_list;

public:
	int m_thread_count;
	HttpServerImp(HttpServer* server, int index, int thread_count) :
		m_base(NULL), m_http(NULL), m_ev(NULL), m_server(server), m_index(index),
		m_task(thread_count, String::format("http%03d_req_tsk", index)),
		Thread(String::format("http%03d_event_trd", index)), 
		m_thread_count(thread_count)
	{
		this->Run();
	}

	virtual int Process()
	{
		m_base = event_base_new();
		if (!m_base)
		{
			LOGERR("%s: couldn't create an event_base.", __FUNCTION__);
			return 1;
		}

		m_http = evhttp_new(m_base);
		if (!m_http)
		{
			LOGERR("%s: couldn't create evhttp.", __FUNCTION__);
			return 2;
		}

		m_ev = event_new(m_base, -1, EV_PERSIST | EV_READ, http_reply_cb, this);
		if (!m_ev)
		{
			LOGERR("%s: couldn't create event.", __FUNCTION__);
			return 3;
		}
		if (0 != event_add(m_ev, NULL))
		{
			LOGERR("%s: couldn't add event.", __FUNCTION__);
			return 4;
		}

#if (COMM_FEATURE_SSL==1)
		SSL_CTX* server_ctx = NULL;
		server_ctx = SSLInit(m_server->m_crtfile, m_server->m_keyfile);
		MYLIB_ASSERT_NOTNULL(server_ctx);
		evhttp_set_bevcb(m_http, http_bevcb, server_ctx);
#endif

		evhttp_set_timeout(m_http, m_server->m_timeout_seconds);
		evhttp_set_gencb(m_http, http_request_cb, this);

		MYLIB_ASSERT(INVALID_SOCKET != m_server->m_sock);
		if (0 != evhttp_accept_socket(m_http, m_server->m_sock))
		{
			LOGERR("%s: couldn't accept socket.", __FUNCTION__);
			return 5;
		}

		if (0 != event_base_dispatch(m_base))
		{
			LOGERR("%s: couldn't dispatch event_base.", __FUNCTION__);
			return 6;
		}

		return 0;
	}

	void Stop()
	{
		if (NULL != m_base)
		{
			MYLIB_ASSERT(0 == event_base_loopexit(m_base, NULL));
			this->Join();
		}
		m_task.Stop();

		if (NULL != m_http)
		{
			evhttp_free(m_http);
			m_http = NULL;
		}
		if (NULL != m_base)
		{
			event_base_free(m_base);
			m_base = NULL;
		}
		if (NULL != m_ev)
		{
			event_free(m_ev);
			m_ev = NULL;
		}
	}

	void ScheduleRequest(struct evhttp_request *req)
	{
		RequestItem* item = new RequestItem(req, this);
		MYLIB_ASSERT_NOTNULL(item);
		m_task.Schedule(item);
	}

	void PostReplyRequest(struct evhttp_request *req, bool pageFound)
	{
		ThreadLockGuard guard(&m_lock_list);
		m_reply_list.push_back(std::make_pair(req, pageFound));
		event_active(m_ev, EV_READ, 0);
	}

	void DoReplyRequest()
	{
		ThreadLockGuard guard(&m_lock_list);
		//LOGINFO("%s:%d: %d.", __FUNCTION__, __LINE__, m_reply_list.size());
		for (ReplyList::iterator itor = m_reply_list.begin(); itor != m_reply_list.end(); itor++)
		{
			evhttp_request *req = itor->first;
			bool pageFound = itor->second;
			if (pageFound)
			{
				evhttp_send_reply(req, 200, "OK", NULL);
			}
			else
			{
				evhttp_send_reply(req, 404, "not found", NULL);
			}
		}

		m_reply_list.clear();
	}

};



static void http_request_cb(struct evhttp_request *req, void *arg)
{
	HttpServerImp* instance = (HttpServerImp*)arg;
	if (instance->m_thread_count > 1)
	{
		instance->ScheduleRequest(req);
		return;
	}

	HttpServerImp::RequestItem item(req, instance, true);
	item.Do();
}

static void http_reply_cb(evutil_socket_t sock, short which, void *arg)
{
	HttpServerImp* instance = (HttpServerImp*)arg;
	instance->DoReplyRequest();
}


HttpServer::HttpServer(HttpRequestHandler* handler,
	int listenPort,
	const char* listenIP,
	int timeoutSeconds,
	int threadCountPerCore,
	int coreCount,
	const char* contentType,
	const char* crtfile, const char* keyfile)
	: m_handler(handler),
	m_port(listenPort), m_ip(listenIP),
	m_timeout_seconds(timeoutSeconds),
	m_thread_count_per_core(threadCountPerCore),
	m_core_count(coreCount),
	m_content_type(contentType),
	m_crtfile(crtfile), m_keyfile(keyfile),
	m_sock(INVALID_SOCKET)
{
	MYLIB_ASSERT_NOTNULL(m_handler);
#ifdef _WIN32
	WSADATA WSAData;
	WSAStartup(0x0202, &WSAData);
#else
	MYLIB_ASSERT(SIG_ERR != signal(SIGPIPE, SIG_IGN));
#endif

#ifdef _MSC_VER
	MYLIB_ASSERT(0 == evthread_use_windows_threads());
#else
	MYLIB_ASSERT(0 == evthread_use_pthreads());
#endif
}

HttpServer::~HttpServer(void)
{
	Stop();
	if (NULL != m_handler)
	{
		m_handler->Release();
		m_handler = NULL;
	}
}

bool HttpServer::Start()
{
	ThreadLockGuard guard(&m_lock);
	if (!BindSocket())
		return false;
	for (int i = 0; i < m_core_count; i++)
	{
		HttpServerImp* instance = new HttpServerImp(this, i + 1, MYLIB_MAX(1, m_thread_count_per_core));
		MYLIB_ASSERT_NOTNULL(instance);
		m_instance_list.push_back(instance);
	}
	return true;
}

void HttpServer::Stop()
{
	ThreadLockGuard guard(&m_lock);
	for (std::list<void*>::iterator itor = m_instance_list.begin(); itor != m_instance_list.end(); itor++)
	{
		HttpServerImp* instance = (HttpServerImp*)(*itor);
		instance->Stop();
		delete instance;
	}
	m_instance_list.clear();
	CloseSocket();
}

bool HttpServer::BindSocket()
{
	COMM_SOCKET sock = INVALID_SOCKET;
	do
	{
		sock = ::socket(AF_INET, SOCK_STREAM, 0);
		if (INVALID_SOCKET == sock)
		{
			LOGFUNCERRORWSA(socket);
			break;
		}

		if (0 != evutil_make_listen_socket_reuseable(sock))
		{
			LOGFUNCERRORWSA(evutil_make_listen_socket_reuseable);
			break;
		}

		if (0 != evutil_make_socket_nonblocking(sock))
		{
			LOGFUNCERRORWSA(evutil_make_socket_nonblocking);
			break;
		}

		SocketAddress address(m_port, m_ip);
		if (INVALID_SOCKET == ::bind(sock, (sockaddr*)address, sizeof(struct sockaddr)))
		{
			LOGFUNCERRORWSA(bind);
			break;
		}

		if (INVALID_SOCKET == ::listen(sock, SOMAXCONN))
		{
			LOGFUNCERRORWSA(listen);
			break;
		}

		m_sock = sock;
		LOGINFO("http-server(%s) %d core * %d threads is running...", address.ToString().get(), m_core_count, m_thread_count_per_core);
		return true;

	} while (false);

	if (INVALID_SOCKET != sock)
	{
		evutil_closesocket(sock);
	}
	return false;
}

void HttpServer::CloseSocket()
{
	if (INVALID_SOCKET != m_sock)
	{
		evutil_closesocket(m_sock);
		m_sock = INVALID_SOCKET;
	}
}



#endif